package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.DimAsyncFunction;

import com.atguigu.edu.bean.TradeCourseOrderBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import com.google.inject.internal.cglib.proxy.$Callback;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO　2 设置状态后端
                 /*
                 env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                 env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                 env.setStateBackend(new HashMapStateBackend());
                 env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                 System.setProperty("HADOOP_USER_NAME", "atguigu");
                  */
        //TODO 3 读取kafka DWD层对应主题的数据
        String topicName = "dwd_trade_order_detail";
        String groupID = "dws_trade_course_order_window";
        DataStreamSource<String> orderStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));
        //TODO 4 转化结构
        //TODO 5 过滤掉不完整的数据
        SingleOutputStreamOperator<JSONObject> jsonObjStream = orderStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String userId = jsonObject.getString("user_id");
                if (userId != null) {
                    collector.collect(jsonObject);
                }
            }
        });
        jsonObjStream.print("json>>>>>>>>>");
        //TODO 6 按照订单id分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject object) throws Exception {
                return object.getString("id");
            }
        });
        keyedStream.print("key>>>>>>>");
        //TODO 7 根据课程id来聚合再开窗
        SingleOutputStreamOperator<TradeCourseOrderBean> beanStream = keyedStream.map(new MapFunction<JSONObject,
                TradeCourseOrderBean>() {
            @Override
            public TradeCourseOrderBean map(JSONObject object) throws Exception {
                return TradeCourseOrderBean.builder()
                        .courseId(object.getString("course_id"))
                        .courseName(object.getString("course_name"))
                        .finalAmount(object.getString("final_amount"))
                        .ts(object.getLong("od_ts") * 1000L)
                        .build();
            }
        });
        SingleOutputStreamOperator<TradeCourseOrderBean> beanWithWatermarkStream =
                beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeCourseOrderBean>() {
            @Override
            public long extractTimestamp(TradeCourseOrderBean tradeCourseOrderBeam, long l) {
                return tradeCourseOrderBeam.getTs();
            }
        }));
        SingleOutputStreamOperator<TradeCourseOrderBean> reduceStream =
                beanWithWatermarkStream.keyBy(new KeySelector<TradeCourseOrderBean, String>() {
            @Override
            public String getKey(TradeCourseOrderBean tradeCourseOrderBeam) throws Exception {
                return tradeCourseOrderBeam.getCourseId();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean reduce(TradeCourseOrderBean tradeCourseOrderBeam,
                                                       TradeCourseOrderBean t1) throws Exception {
                        tradeCourseOrderBeam.setFinalAmount(tradeCourseOrderBeam.getFinalAmount() + t1.getFinalAmount());
                        return tradeCourseOrderBeam;
                    }
                }, new WindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeCourseOrderBean> iterable,
                                      Collector<TradeCourseOrderBean> collector) throws Exception {
                        TradeCourseOrderBean tradeCourseOrderBeam = iterable.iterator().next();
                        tradeCourseOrderBeam.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                        tradeCourseOrderBeam.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                        tradeCourseOrderBeam.setTs(System.currentTimeMillis());
                        collector.collect(tradeCourseOrderBeam);
                    }
                });

        beanStream.print("bean>>>>>>");
        //  TODO 8 和DIM层的维度表进行关联
        // 关联course
        SingleOutputStreamOperator<TradeCourseOrderBean> courseStream =
                AsyncDataStream.unorderedWait(reduceStream, new DimAsyncFunction<TradeCourseOrderBean>(
                        "DIM_COURSE_INFO") {
            @Override
            public String getKey(TradeCourseOrderBean input) {
                return input.getCourseId();
            }

            @Override
            public void join(TradeCourseOrderBean input, JSONObject obj) {
                input.setCourseName(obj.getString("courseName"));
                input.setSubjectId(obj.getString("subjectId"));

            }
        }, 100, TimeUnit.SECONDS);
        courseStream.print("course>>>>>");

        //关联subject
        SingleOutputStreamOperator<TradeCourseOrderBean> subjectStream =
                AsyncDataStream.unorderedWait(courseStream, new DimAsyncFunction<TradeCourseOrderBean>(
                        "DIM_BASE_SUBJECT_INFO") {
                    @Override
                    public String getKey(TradeCourseOrderBean input) {
                        return input.getSubjectId();
                    }

                    @Override
                    public void join(TradeCourseOrderBean input, JSONObject obj) {
                        input.setSubjectName(obj.getString("subjectName"));
                        input.setCategoryId(obj.getString("categoryId"));

                    }
                }, 100, TimeUnit.SECONDS);
        subjectStream.print("subject>>>>>");
        //关联category
        SingleOutputStreamOperator<TradeCourseOrderBean> resultStream =
                AsyncDataStream.unorderedWait(subjectStream, new DimAsyncFunction<TradeCourseOrderBean>(
                        "DIM_BASE_CATEGORY_INFO") {
                    @Override
                    public String getKey(TradeCourseOrderBean input) {
                        return input.getCategoryId();
                    }

                    @Override
                    public void join(TradeCourseOrderBean input, JSONObject obj) {
                        input.setCategoryName(obj.getString("categoryName"));

                    }
                }, 100, TimeUnit.SECONDS);

        resultStream.print("result>>>>>>>>>");
        //TODO 9 写出数据到ClickHouse中
        /*resultStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_trade_course_order_window values " +
                "(?,?,?,?,?,?,?,?,?,?)"));*/

        //TODO 10 执行任务
        env.execute(groupID);

    }
}
